package com.zhisheng.sql.blink.stream.example;

import org.apache.commons.math3.util.IntegerSequence;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;


public class TestSql {
    public static void main(String[] args) throws Exception {
        // 1. 创建 Flink 流执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        EnvironmentSettings settings = EnvironmentSettings.newInstance().inStreamingMode().build();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);

        // 2. 创建 Kafka 数据源表（学生表）
        tableEnv.executeSql("CREATE TABLE student (" + "  id INT PRIMARY KEY," + "  name STRING," + "  age INT," + "  t_id INT" + ") WITH (" + "  'connector' = 'kafka'," + "  'topic' = 'mysql.test_sm.student'," + "  " +
                "'properties.bootstrap.servers' = '192.168.8.61:9092,192.168.8.61:9093,192.168.8.61:9094'," + " " +
                " 'properties.group.id' = 'test_2'," + " " +
                " 'scan.startup.mode' = 'earliest-offset'," +
                "  'format' = 'debezium-json'," +
                " 'debezium-json.schema-include' = 'true'," +
                "  'debezium-json.ignore-parse-errors' = 'true'" + ")");

        // 3. 创建 Kafka 数据源表（老师表）
        tableEnv.executeSql("CREATE TABLE teacher (" + "  id INT PRIMARY KEY," + "  name STRING" + ") WITH (" + "  'connector' = 'kafka'," + "  'topic' = 'mysql.test_sm.teacher'," + "  'properties.bootstrap.servers' = '192.168.8.61:9092,192.168.8.61:9093,192.168.8.61:9094'," + "  'properties.group.id' = 'test_2'," + "  'properties.auto.offset.reset' = 'earliest'," + "  'scan.startup.mode' = 'earliest-offset'," + " 'debezium-json.schema-include' = 'true'," + "  'format' = 'debezium-json'" + ")");

        // 创建关联视图（使用 Temporal Join 确保总能看到老师最新信息）
        tableEnv.executeSql(
                "CREATE TEMPORARY VIEW joined_students AS " +
                        "SELECT s.id AS student_id, s.t_id, t.name AS teacher_name " +
                        "FROM student s " +
                        "JOIN teacher t " +
                        "ON s.t_id = t.id"
        );
        // 计算每个老师的学生数并实时更新
        tableEnv.executeSql(
                "CREATE TEMPORARY VIEW teacher_counts AS " +
                        "SELECT t_id, teacher_name, COUNT(student_id) AS student_count " +
                        "FROM joined_students " +
                        "GROUP BY t_id, teacher_name"
        );

        // 实时计算 Top5 排行榜（处理更新）
        Table resultTable = tableEnv.sqlQuery(
                "SELECT t_id, teacher_name, student_count, row_num " +
                        "FROM (" +
                        "  SELECT *, ROW_NUMBER() OVER (ORDER BY student_count DESC) AS row_num " +
                        "  FROM teacher_counts" +
                        ") WHERE row_num <= 10"
        );


        // 配置 JDBC Sink 将数据实时写入 MySQL
        String mysqlUrl = "jdbc:mysql://192.168.8.61:3306/test_sm";
        String mysqlUser = "root";
        String mysqlPassword = "Cyx1018!";

        // 设置 JDBC Sink
        SinkFunction<Row> jdbcSink = JdbcSink.sink(
                "INSERT INTO teacher_rank (teacher_id,teacher_name, student_count, rank_number) VALUES (?, ?, ?,?) " +
                        "ON DUPLICATE KEY UPDATE student_count = VALUES(student_count), teacher_name = values(teacher_name),rank_number = VALUES(rank_number)",
                (ps, t) -> {
                    ps.setInt(1, Integer.valueOf( t.getField(0).toString())); // teacher_id
                    ps.setString(2, String.valueOf( t.getField(1)));  // student_count
                    ps.setInt(3, Integer.valueOf( t.getField(2).toString()));  // student_count
                    ps.setInt(4, Integer.valueOf( t.getField(3).toString())); // rank
                },
                new JdbcExecutionOptions.Builder()
                        .withBatchSize(1) // 每条数据单独插入
                        .withMaxRetries(3) // 重试次数
                        .build(),
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withUrl(mysqlUrl)
                        .withDriverName("com.mysql.cj.jdbc.Driver")
                        .withUsername(mysqlUser)
                        .withPassword(mysqlPassword)
                        .build()
        );

        // 将结果转换为 DataStream
        DataStream<Row> resultStream = tableEnv.toChangelogStream(resultTable);
        // 将结果流写入 MySQL
        resultStream.addSink(jdbcSink);

        // 打印变更日志流（包含排名更新）
        tableEnv.toChangelogStream(resultTable).print("Top5 Teachers:");
        env.execute("Teacher Student Count Job");
    }
}